package vip.meeet.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class ManualConsumerC {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("group.id", "ok");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset","earliest");
        props.put("max.poll.records",1000);
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        Duration duration = Duration.of(100, ChronoUnit.MILLIS);
        consumer.subscribe(Arrays.asList("test"));
        int count = 0;
        Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<TopicPartition,OffsetAndMetadata>();
        try {
            while (true){
                ConsumerRecords<String,String> records = consumer.poll(duration);
                for(ConsumerRecord<String ,String> record : records){
                    count ++;
                    //offset + 1，下次消费者从该偏移量开始拉取消息
                    currentOffsets.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset()+1,"no"));
                    if ((count / 10 == 1) && (count % 10 == 0)){
                        System.out.println(count);
                        consumer.commitSync(currentOffsets);
                    }
                    System.out.println(record.topic() + "," + record.partition() + "," + record.offset() + "," + record.key() + "," + record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}
